package com.zyk.scaffold.mq.rocket.business.order.producers;


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.List;

@Slf4j
@Component
public class OrderlyProducer {

	private Producer producer;
	private ClientServiceProvider provider;

	private OrderlyProducer() throws ClientException {
		// 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
		String endpoint = "localhost:8081";
		// 消息发送的目标Topic名称，需要提前创建。
		String topic = "TestTopic";
		provider = ClientServiceProvider.loadService();
		ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
		ClientConfiguration configuration = builder.build();
		// 初始化Producer时需要设置通信配置以及预绑定的Topic。
		producer = provider.newProducerBuilder()
				.setTopics(topic)
				.setClientConfiguration(configuration)
				.build();
	}

	public Message newMessageBuilder(String key, String body){
		return provider.newMessageBuilder()
				.setTopic("topic")
				// 设置消息索引键，可根据关键字精确查找某条消息。
				.setKeys("messageKey")
				// 设置消息Tag，用于消费端根据指定Tag过滤消息。
				.setTag("tag")
				// 消息体。
				.setBody("messageBody".getBytes())
				.build();
	}
	
	public void sendOrderlyMessages(List<Message> messages, int messageQueueNumber) {
		for(Message me : messages) {
			try {
				// 发送消息，需要关注发送结果，并捕获失败等异常。
				SendReceipt sendReceipt = producer.send(me);
				log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
			} catch (ClientException e) {
				log.error("Failed to send message", e);
			}
		}
	}

	@PreDestroy
	private void destory(){
		if(producer != null){
			try {
				// 如果不需要再使用 producer，可关闭该实例。
				producer.close();
			} catch (IOException e) {
				log.error("pushConsumer close error", e);
			}
		}
	}
}
